"""
This file contains a class that is a thread for doing I/O on the specified
socket.

This thread is useful because it seems that if the channel does not
continuously read from the socket, the other side may halt execution of
whatever is outputting to the socket which is bad in an expect situation.

It is also useful for writing data, since we do not want the write method to
block when buffering is enabled.

This is easier if we have two threads per channel, but to save on threads i'm
doing it with one per channel.

Copyright (c) 2009 Christopher Hesse
http://code.google.com/p/python-expect/
"""
import threading
import select
import errno
import logging
from expect import is_windows
from expect.exceptions import ChannelClosed

READ_SIZE = 8092
log = logging.getLogger(__name__)


class IOThread(threading.Thread):
    """
    This thread handles I/O for a SocketChannel, it continuously reads
    any data available on the socket and writes any data waiting to be
    written to the socket.
    """
    
    def __init__(self, name, socket, write_queue, write_selectable, 
                 read_queue, read_selectable):
        thread_name = "IOThread for channel '{0}'".format(name)
        super(IOThread, self).__init__(name=thread_name)
        self.channel_name = name
        self.socket = socket
        self.write_queue = write_queue
        self.write_selectable = write_selectable
        self.read_queue = read_queue
        self.read_selectable = read_selectable
        
    def _log_read(self, data):
        log.info(data, extra={'channel': self.channel_name, 'direction': 'read'})

    def _log_write(self, data):
        log.info(data, extra={'channel': self.channel_name, 'direction': 'write'})
    
    def run(self):
        return self.process_io() # pragma: no cover
    
    def process_io(self):
        try:
            self.loop_forever()
        except Exception, e:
            allowed_error_codes = [errno.EBADF]
            if is_windows:
                allowed_error_codes.append(errno.WSAECONNABORTED)
            else:
                allowed_error_codes.append(errno.EPIPE)
                allowed_error_codes.append(errno.ECONNRESET)
            
            if hasattr(e, 'errno'):
                error_code = e.errno
            else:
                error_code = e.args[0]
            
            if error_code in allowed_error_codes:
                # we expect these errors but end the IO thread anyway
                log.debug("IOThread ended")
            else:
                raise
        # put the channel closed event on the read queue
        with self.read_selectable:
            self.read_queue.put_nowait(ChannelClosed)
            self.read_selectable.set()
    
    def loop_forever(self):
        # loop and select on the channel until the channel is closed
        writables_to_select_on = []
        data_to_write = ""
        task_counter = 0
        while True:
            readable, writable, _other = \
                select.select([self.socket, self.write_selectable], 
                              writables_to_select_on, 
                              [])
            
            if len(writable) > 0:
                # this means both that the socket is writable and that we have data
                # that needs to be written
                bytes_written = self.socket.send(data_to_write)
                self._log_write(data_to_write[:bytes_written])
                data_to_write = data_to_write[bytes_written:]
                if len(data_to_write) == 0:
                    # no more data left to write
                    writables_to_select_on = []
                    # mark all items we got off the write_queue to be done
                    for i in xrange(task_counter):
                        self.write_queue.task_done()
                    task_counter = 0
            
            if self.socket in readable:
                # there's data to be read from the socket
                read_data = self.socket.recv(READ_SIZE)
                if read_data == '':
                    log.debug("Got EOF in IOThread")
                    break
                with self.read_selectable:
                    self.read_queue.put_nowait(read_data)
                    self.read_selectable.set()
                self._log_read(read_data)
            
            if self.write_selectable in readable:
                # this means we have data to write and we've already written
                # all our old data
                with self.write_selectable:
                    for i in xrange(self.write_queue.qsize()):
                        data_to_write += self.write_queue.get_nowait()
                        task_counter += 1
                    self.write_selectable.clear()
                writables_to_select_on = [self.socket]
